
package com.rrd.dw.azkaban.flow;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

public class Flow {
    private final String id;
    private int projectId;
    private ArrayList<Node> startNodes = null;
    private ArrayList<Node> endNodes = null;
    private int numLevels = -1;

    private HashMap<String, Node> nodes = new HashMap<>();

    private HashMap<String, Edge> edges = new HashMap<>();
    private HashMap<String, Set<Edge>> outEdges = new HashMap<>();
    private HashMap<String, Set<Edge>> inEdges = new HashMap<>();
    private HashMap<String, FlowProps> flowProps = new HashMap<>();

    private List<String> failureEmail = new ArrayList<>();
    private List<String> successEmail = new ArrayList<>();
    private ArrayList<String> errors;
    private int version = -1;
    private Map<String, Object> metadata = new HashMap<>();

    private boolean isLayedOut = false;

    public Flow(String id) {
        this.id = id;
    }

    public void setVersion(int version) {
        this.version = version;
    }

    public int getVersion() {
        return this.version;
    }

    public void initialize() {
        if (this.startNodes == null) {
            this.startNodes = new ArrayList<>();
            this.endNodes = new ArrayList<>();
            for (Node node : this.nodes.values()) {
                // If it doesn't have any incoming edges, its a start node
                if (!this.inEdges.containsKey(node.getId())) {
                    this.startNodes.add(node);
                }

                // If it doesn't contain any outgoing edges, its an end node.
                if (!this.outEdges.containsKey(node.getId())) {
                    this.endNodes.add(node);
                }
            }

            for (Node node : this.startNodes) {
                node.setLevel(0);
                this.numLevels = 0;
                this.recursiveSetLevels(node);
            }
        }
    }

    private void recursiveSetLevels(Node node) {
        Set<Edge> edges = this.outEdges.get(node.getId());
        if (edges != null) {
            for (Edge edge : edges) {
                Node nextNode = this.nodes.get(edge.getTargetId());
                edge.setSource(node);
                edge.setTarget(nextNode);

                // We pick whichever is higher to get the max distance from root.
                int level = Math.max(node.getLevel() + 1, nextNode.getLevel());
                nextNode.setLevel(level);
                this.numLevels = Math.max(level, this.numLevels);
                this.recursiveSetLevels(nextNode);
            }
        }
    }

    public Node getNode(String nodeId) {
        return this.nodes.get(nodeId);
    }

    public List<String> getSuccessEmails() {
        return this.successEmail;
    }

    public List<String> getFailureEmails() {
        return this.failureEmail;
    }

    public void addSuccessEmails(Collection<String> emails) {
        this.successEmail.addAll(emails);
    }

    public void addFailureEmails(Collection<String> emails) {
        this.failureEmail.addAll(emails);
    }

    public int getNumLevels() {
        return this.numLevels;
    }

    public List<Node> getStartNodes() {
        return this.startNodes;
    }

    public List<Node> getEndNodes() {
        return this.endNodes;
    }

    public Set<Edge> getInEdges(String id) {
        return this.inEdges.get(id);
    }

    public Set<Edge> getOutEdges(String id) {
        return this.outEdges.get(id);
    }

    public void addAllNodes(Collection<Node> nodes) {
        for (Node node : nodes) {
            this.addNode(node);
        }
    }

    public void addNode(Node node) {
        this.nodes.put(node.getId(), node);
    }

    public void addAllFlowProperties(Collection<FlowProps> props) {
        for (FlowProps prop : props) {
            this.flowProps.put(prop.getSource(), prop);
        }
    }

    public String getId() {
        return this.id;
    }

    public void addError(String error) {
        if (this.errors == null) {
            this.errors = new ArrayList<>();
        }

        this.errors.add(error);
    }

    public List<String> getErrors() {
        return this.errors;
    }

    public boolean hasErrors() {
        return this.errors != null && !this.errors.isEmpty();
    }

    public Collection<Node> getNodes() {
        return this.nodes.values();
    }

    public Collection<Edge> getEdges() {
        return this.edges.values();
    }

    public void addAllEdges(Collection<Edge> edges) {
        for (Edge edge : edges) {
            this.addEdge(edge);
        }
    }

    public void addEdge(Edge edge) {
        String source = edge.getSourceId();
        String target = edge.getTargetId();

        if (edge.hasError()) {
            this.addError("Error on " + edge.getId() + ". " + edge.getError());
        }

        Set<Edge> sourceSet = this.getEdgeSet(this.outEdges, source);
        sourceSet.add(edge);

        Set<Edge> targetSet = this.getEdgeSet(this.inEdges, target);
        targetSet.add(edge);

        this.edges.put(edge.getId(), edge);
    }

    private Set<Edge> getEdgeSet(HashMap<String, Set<Edge>> map, String id) {
        Set<Edge> edges = map.get(id);
        if (edges == null) {
            edges = new HashSet<>();
            map.put(id, edges);
        }

        return edges;
    }

    public Map<String, Object> toObject() {
        HashMap<String, Object> flowObj = new HashMap<>();
        flowObj.put("type", "flow");
        flowObj.put("id", this.getId());
        flowObj.put("project.id", this.projectId);
        flowObj.put("version", this.version);
        flowObj.put("props", this.objectizeProperties());
        flowObj.put("nodes", this.objectizeNodes());
        flowObj.put("edges", this.objectizeEdges());
        flowObj.put("failure.email", this.failureEmail);
        flowObj.put("success.email", this.successEmail);
        flowObj.put("layedout", this.isLayedOut);
        if (this.errors != null) {
            flowObj.put("errors", this.errors);
        }

        if (this.metadata != null) {
            flowObj.put("metadata", this.metadata);
        }

        return flowObj;
    }

    private List<Object> objectizeProperties() {
        ArrayList<Object> result = new ArrayList<>();
        for (FlowProps props : this.flowProps.values()) {
            Object objProps = props.toObject();
            result.add(objProps);
        }

        return result;
    }

    private List<Object> objectizeNodes() {
        ArrayList<Object> result = new ArrayList<>();
        for (Node node : this.getNodes()) {
            Object nodeObj = node.toObject();
            result.add(nodeObj);
        }

        return result;
    }

    private List<Object> objectizeEdges() {
        ArrayList<Object> result = new ArrayList<>();
        for (Edge edge : this.getEdges()) {
            Object edgeObj = edge.toObject();
            result.add(edgeObj);
        }

        return result;
    }

    @SuppressWarnings("unchecked")
    public static Flow flowFromObject(Object object) {
        Map<String, Object> flowObject = (Map<String, Object>) object;

        String id = (String) flowObject.get("id");
        Boolean layedout = (Boolean) flowObject.get("layedout");
        Flow flow = new Flow(id);
        if (layedout != null) {
            flow.setLayedOut(layedout);
        }
        int projId = (Integer) flowObject.get("project.id");
        flow.setProjectId(projId);

        int version = (Integer) flowObject.get("version");
        flow.setVersion(version);

        // Loading projects
        List<Object> propertiesList = (List<Object>) flowObject.get("props");
        Map<String, FlowProps> properties = Flow
            .loadPropertiesFromObject(propertiesList);
        flow.addAllFlowProperties(properties.values());

        // Loading nodes
        List<Object> nodeList = (List<Object>) flowObject.get("nodes");
        Map<String, Node> nodes = Flow.loadNodesFromObjects(nodeList);
        flow.addAllNodes(nodes.values());

        // Loading edges
        List<Object> edgeList = (List<Object>) flowObject.get("edges");
        List<Edge> edges = Flow.loadEdgeFromObjects(edgeList, nodes);
        flow.addAllEdges(edges);

        Map<String, Object> metadata = (Map<String, Object>) flowObject
            .get("metadata");

        if (metadata != null) {
            flow.setMetadata(metadata);
        }

        flow.failureEmail = (List<String>) flowObject.get("failure.email");
        flow.successEmail = (List<String>) flowObject.get("success.email");
        return flow;
    }

    private static Map<String, Node>
            loadNodesFromObjects(List<Object> nodeList) {
        Map<String, Node> nodeMap = new HashMap<>();

        for (Object obj : nodeList) {
            Node node = Node.fromObject(obj);
            nodeMap.put(node.getId(), node);
        }

        return nodeMap;
    }

    private static List<Edge> loadEdgeFromObjects(List<Object> edgeList,
            Map<String, Node> nodes) {
        List<Edge> edgeResult = new ArrayList<>();

        for (Object obj : edgeList) {
            Edge edge = Edge.fromObject(obj);
            edgeResult.add(edge);
        }

        return edgeResult;
    }

    private static Map<String, FlowProps>
            loadPropertiesFromObject(List<Object> propertyObjectList) {
        Map<String, FlowProps> properties = new HashMap<>();

        for (Object propObj : propertyObjectList) {
            FlowProps prop = FlowProps.fromObject(propObj);
            properties.put(prop.getSource(), prop);
        }

        return properties;
    }

    public boolean isLayedOut() {
        return this.isLayedOut;
    }

    public Map<String, Object> getMetadata() {
        if (this.metadata == null) {
            this.metadata = new HashMap<>();
        }
        return this.metadata;
    }

    public void setMetadata(Map<String, Object> metadata) {
        this.metadata = metadata;
    }

    public void setLayedOut(boolean layedOut) {
        this.isLayedOut = layedOut;
    }

    public Map<String, Node> getNodeMap() {
        return this.nodes;
    }

    public Map<String, Set<Edge>> getOutEdgeMap() {
        return this.outEdges;
    }

    public Map<String, Set<Edge>> getInEdgeMap() {
        return this.inEdges;
    }

    public FlowProps getFlowProps(String propSource) {
        return this.flowProps.get(propSource);
    }

    public Map<String, FlowProps> getAllFlowProps() {
        return this.flowProps;
    }

    public int getProjectId() {
        return this.projectId;
    }

    public void setProjectId(int projectId) {
        this.projectId = projectId;
    }
}
